// Copyright 2013 Google Inc. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package dfa.axis.v1_19.matchtables;

import static dfa.axis.v1_19.matchtables.CreativeTypes.SoapCreativeType;

import com.google.api.ads.dfa.axis.factory.DfaServices;
import com.google.api.ads.dfa.axis.v1_19.AdBase;
import com.google.api.ads.dfa.axis.v1_19.AdRecordSet;
import com.google.api.ads.dfa.axis.v1_19.AdRemote;
import com.google.api.ads.dfa.axis.v1_19.AdSearchCriteria;
import com.google.api.ads.dfa.axis.v1_19.AdType;
import com.google.api.ads.dfa.axis.v1_19.ApiException;
import com.google.api.ads.dfa.axis.v1_19.Campaign;
import com.google.api.ads.dfa.axis.v1_19.CampaignRecordSet;
import com.google.api.ads.dfa.axis.v1_19.CampaignRemote;
import com.google.api.ads.dfa.axis.v1_19.CampaignSearchCriteria;
import com.google.api.ads.dfa.axis.v1_19.ClickThroughUrl;
import com.google.api.ads.dfa.axis.v1_19.ClickTracker;
import com.google.api.ads.dfa.axis.v1_19.CreativeAd;
import com.google.api.ads.dfa.axis.v1_19.CreativeAssignment;
import com.google.api.ads.dfa.axis.v1_19.CreativeBase;
import com.google.api.ads.dfa.axis.v1_19.CreativeField;
import com.google.api.ads.dfa.axis.v1_19.CreativeFieldAssignment;
import com.google.api.ads.dfa.axis.v1_19.CreativeFieldRecordSet;
import com.google.api.ads.dfa.axis.v1_19.CreativeFieldRemote;
import com.google.api.ads.dfa.axis.v1_19.CreativeFieldSearchCriteria;
import com.google.api.ads.dfa.axis.v1_19.CreativeFieldValue;
import com.google.api.ads.dfa.axis.v1_19.CreativeFieldValueRecordSet;
import com.google.api.ads.dfa.axis.v1_19.CreativeFieldValueSearchCriteria;
import com.google.api.ads.dfa.axis.v1_19.CreativeGroup;
import com.google.api.ads.dfa.axis.v1_19.CreativeGroupRecordSet;
import com.google.api.ads.dfa.axis.v1_19.CreativeGroupRemote;
import com.google.api.ads.dfa.axis.v1_19.CreativeGroupSearchCriteria;
import com.google.api.ads.dfa.axis.v1_19.CreativeRecordSet;
import com.google.api.ads.dfa.axis.v1_19.CreativeRemote;
import com.google.api.ads.dfa.axis.v1_19.CreativeRenderingRequest;
import com.google.api.ads.dfa.axis.v1_19.CreativeRenderingResult;
import com.google.api.ads.dfa.axis.v1_19.CreativeSearchCriteria;
import com.google.api.ads.dfa.axis.v1_19.DefaultAd;
import com.google.api.ads.dfa.axis.v1_19.ImageCreativeBase;
import com.google.api.ads.dfa.axis.v1_19.LandingPage;
import com.google.api.ads.dfa.axis.v1_19.LandingPageRecordSet;
import com.google.api.ads.dfa.axis.v1_19.MobileAd;
import com.google.api.ads.dfa.axis.v1_19.PlacementAssignment;
import com.google.api.ads.dfa.axis.v1_19.RedirectCreativeBase;
import com.google.api.ads.dfa.axis.v1_19.RichMediaCreativeBase;
import com.google.api.ads.dfa.axis.v1_19.RichMediaEventBase;
import com.google.api.ads.dfa.axis.v1_19.RotationGroup;
import com.google.api.ads.dfa.axis.v1_19.Size;
import com.google.api.ads.dfa.axis.v1_19.SizeRecordSet;
import com.google.api.ads.dfa.axis.v1_19.SizeRemote;
import com.google.api.ads.dfa.axis.v1_19.SizeSearchCriteria;
import com.google.api.ads.dfa.axis.v1_19.TargetableAdBase;
import com.google.api.ads.dfa.axis.v1_19.TrackingAd;
import com.google.api.ads.dfa.lib.client.DfaSession;

import dfa.axis.v1_19.matchtables.CreativeTypes.CreativeRotationType;

import java.rmi.RemoteException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * Implementation of LookupObject that will generate Campaign, Ad, and Creative match tables.
 *
 * @author Moshe Kahan
 */
class CampaignAdCreativeLookup implements LookupObject {

  // Main aggregation maps
  // Ads per campaign
  private Map<Long, Set<Long>> campaignToAdIdsMap;
  private Map<Long, Set<Long>> advertiserToCampaignIdsMap;
  private Map<String, Set<String>> adToRenderingIdMap;
  private Map<String, Set<String>> advertiserToMetEventIdMap;
  private Set<Long> renderingIdsSet;

  // Secondary lookup maps
  private Map<Long, String> adTypeMap;
  private Map<Long, String> sizeMap;
  private Map<Long, Long> campaignDefaultLandingPageMap;

  // DT columns that this class is interested in
  private int adColumnIndex;
  private int advertiserColumnIndex;
  private int creativeColumnIndex;
  private int campaignColumnIndex;
  private int orderIdColumnIndex;
  private int metEventColumnIndex;

  // MatchTable settings and SOAP remote objects
  private TableSettings tableSettings;
  private AdRemote adService;
  private SizeRemote sizeService;
  private CampaignRemote campaignService;
  private CreativeRemote creativeService;
  private CreativeFieldRemote creativeFieldeService;
  private CreativeGroupRemote creativeGroupService;

  // Progress information
  private int campaignCount;
  private int campaignTotal;

  /**
   * Constructor.  Besides the DFA service/session objects needed to create SOAP remote objects to
   * the API, constructor also accepts matchTableTypes to determine if both Ad and Creative match
   * tables should be generated.
   *
   * @param dfaServices DFA services object needed to create SOAP remotes to the API
   * @param session DFA session object needed to create SOAP remotes to the API
   * @param tableSettings match table application configuration used as a MatchTable object factory
   */
  CampaignAdCreativeLookup(DfaServices dfaServices, DfaSession session,
      TableSettings tableSettings) {
    // Request the advertiser service from the service client factory.
    campaignToAdIdsMap = new HashMap<Long, Set<Long>>();
    advertiserToCampaignIdsMap = new HashMap<Long, Set<Long>>();
    adToRenderingIdMap = new HashMap<String, Set<String>>();
    advertiserToMetEventIdMap = new HashMap<String, Set<String>>();
    renderingIdsSet = new HashSet<Long>();

    adTypeMap = new HashMap<Long, String>();
    sizeMap = new HashMap<Long, String>();
    campaignDefaultLandingPageMap = new HashMap<Long, Long>();

    adService = dfaServices.get(session, AdRemote.class);
    sizeService = dfaServices.get(session, SizeRemote.class);
    campaignService = dfaServices.get(session, CampaignRemote.class);
    creativeService = dfaServices.get(session, CreativeRemote.class);
    creativeFieldeService = dfaServices.get(session, CreativeFieldRemote.class);
    creativeGroupService = dfaServices.get(session, CreativeGroupRemote.class);
    this.tableSettings = tableSettings;

    campaignCount = 0;
    campaignTotal = 0;

    adColumnIndex = -1;
    advertiserColumnIndex = -1;
    creativeColumnIndex = -1;
    campaignColumnIndex = -1;
    orderIdColumnIndex = -1;
    metEventColumnIndex = -1;
  }

  /**
   * Called when LookupObjectManager is parsing the DT file header. Method stores index
   * of ad, advertiser, campaign, and creative columns.
   *
   * @param index in DT file columns
   * @param header name of this DT file field/column
   */
  public void addHeader(int index, String header) {
    if (LookupField.AdId.getFieldName().equals(header)) {
      adColumnIndex = index;
    } else if (LookupField.AdvertiserId.getFieldName().equals(header)) {
      advertiserColumnIndex = index;
    } else if (LookupField.BuyId.getFieldName().equals(header)) {
      campaignColumnIndex = index;
    } else if (LookupField.OrderId.getFieldName().equals(header)) {
      orderIdColumnIndex = index;
    } else if (LookupField.CreativeId.getFieldName().equals(header) && creativeColumnIndex < 0) {
      // Only use CreativeID if RenderingID is not present
      // since RM DT's RenderingID == Non-RM DT's CreativeID.
      creativeColumnIndex = index;
    } else if (LookupField.RenderingId.getFieldName().equals(header)) {
      creativeColumnIndex = index;
    } else if (LookupField.MetEventId.getFieldName().equals(header)) {
      metEventColumnIndex = index;
    }
  }

  /**
   * Called to clear the DT header information' usually because you're processing a new file.
   */
  public void clearHeaders() {
    adColumnIndex = -1;
    advertiserColumnIndex = -1;
    creativeColumnIndex = -1;
    campaignColumnIndex = -1;
    orderIdColumnIndex = -1;
    metEventColumnIndex = -1;
  }

  /**
   * Called after DT file line has been tokenized into array of Strings.  Aggregate over 4 column
   * of interest.  Only aggregate creative id if Creative match table is to be generated. Create 
   * grouping of ads IDs per campaign to make it easier for later SOAP API lookups.
   *
   * @param values array of tokenized Strings of DT file line
   */
  public void addValues(String[] values) {
    // Can only parse if Ad-ID, Advertiser-ID, and Buy-ID are present.
    if (advertiserColumnIndex > -1) {
      String advertiserId = values[advertiserColumnIndex];
      if (campaignColumnIndex > -1 && isValid(advertiserId)) {
        String campaignId = values[campaignColumnIndex];
        if (!isValid(campaignId) && orderIdColumnIndex > -1) {
          campaignId = values[orderIdColumnIndex];
        }
        if (isValid(campaignId)) {
          long numericCampaignId = Long.valueOf(campaignId);
          addToSetToMap(advertiserToCampaignIdsMap, Long.valueOf(advertiserId), numericCampaignId);
          if (adColumnIndex > -1) {
            String adId = values[adColumnIndex];
            if (isValid(adId)) {
              addToSetToMap(campaignToAdIdsMap, numericCampaignId, Long.valueOf(adId));
              if (creativeColumnIndex > -1) {
                String renderingId = values[creativeColumnIndex];
                if (isValid(renderingId)) {
                  addToSetToMap(adToRenderingIdMap, adId, renderingId);
                  renderingIdsSet.add(Long.valueOf(renderingId));
                  if (metEventColumnIndex > -1) {
                    String metEventId = values[metEventColumnIndex];
                    if (isValid(metEventId)) {
                      addToSetToMap(advertiserToMetEventIdMap, advertiserId, metEventId);
                    }
                  }
                }
              }
            }
          }
        }
      }
    }
  }

  private boolean isValid(String value) {
    return (value != null && !value.isEmpty() && !value.equals("0"));
  }

  // Utility method to try to add to Strings Set entry in map; creates Set if doesn't exist.
  private <T1, T2> void addToSetToMap(Map<T1, Set<T2>> mapOfStringToSet, T1 key, T2 value) {
    Set<T2> setOfStrings = mapOfStringToSet.get(key);
    if (setOfStrings == null) {
      setOfStrings = new HashSet<T2>();
      mapOfStringToSet.put(key, setOfStrings);
    }
    setOfStrings.add(value);
  }

  /**
   * Called when parsing/aggregation is complete. Will lookup via SOAP all unique DFA entries
   * in the DT file and pass data to MatchTable class.
   *
   * @param outputPath location of match table to be generated.
   */
  public void generateMatchTable(String outputPath)
      throws MatchTableFileException, DfaSoapException {
    List<MatchTable> matchTableList = new ArrayList<MatchTable>();
    MatchTable adMatchTable =
        createMatchTable(MatchTable.Type.AD, matchTableList);
    MatchTable adPageAssignmentsMatchTable =
        createMatchTable(MatchTable.Type.AD_PAGE_ASSIGNMENTS, matchTableList);
    MatchTable creativeMatchTable =
        createMatchTable(MatchTable.Type.CREATIVE, matchTableList);
    MatchTable creativeAssignmentMatchTable =
        createMatchTable(MatchTable.Type.CREATIVE_AD_ASSIGNMENTS, matchTableList);
    MatchTable keyvalMatchTable =
        createMatchTable(MatchTable.Type.KEYVAL, matchTableList);
    MatchTable creativeCustomFieldMatchTable =
        createMatchTable(MatchTable.Type.CUSTOM_CREATIVE_FIELDS, matchTableList);
    MatchTable campaignMatchTable =
        createMatchTable(MatchTable.Type.CAMPAIGN, matchTableList);
    MatchTable metEventMatchTable = null;

    for (MatchTable matchTable : matchTableList) {
      matchTable.open(outputPath);
    }

    generateCampaignMatchTable(campaignToAdIdsMap.keySet(), campaignMatchTable);
    if (campaignMatchTable != null) {
      campaignMatchTable.flush();
    }

    // If we have more than 1 than we know it's not just campaignMatchTable; or if
    // the 1 table we have is not just campaignMatchTable
    if (matchTableList.size() > 1 || !matchTableList.contains(campaignMatchTable)) {
      MatchTable.printProgress(adMatchTable, adPageAssignmentsMatchTable, creativeMatchTable,
          creativeAssignmentMatchTable, keyvalMatchTable, creativeCustomFieldMatchTable,
          metEventMatchTable);

      // Compute total number of campaigns.
      campaignCount = 0;
      campaignTotal = 0;
      for (Set<Long> campaignSet : advertiserToCampaignIdsMap.values()) {
        campaignTotal += campaignSet.size();
      }

      for (Map.Entry<Long, Set<Long>> entry : advertiserToCampaignIdsMap.entrySet()) {
        generateAdsInAdvertiserMatchTable(entry.getKey(), entry.getValue(), adMatchTable,
            adPageAssignmentsMatchTable, creativeMatchTable, creativeAssignmentMatchTable,
            keyvalMatchTable, creativeCustomFieldMatchTable, metEventMatchTable);
        for (MatchTable matchTable : matchTableList) {
          matchTable.flush();
        }
      }
      System.out.printf("\r[%.1f%% completed]\n", 100.0);
    }

    for (MatchTable matchTable : matchTableList) {
      matchTable.close();
    }
  }

  // Utility method to try to use factory to create match table, only add to List if not null.
  private MatchTable createMatchTable(MatchTable.Type type, List<MatchTable> matchTableList) {
    MatchTable table = tableSettings.createMatchTable(type);
    if (table != null) {
      matchTableList.add(table);
    }
    return table;
  }

  // Generates all campaign lookups and match table rows for all Buy-IDs.
  private void generateCampaignMatchTable(Set<Long> campaignIds, MatchTable campaignMatchTable)
      throws DfaSoapException {

    MatchTable.printProgress(campaignMatchTable);
    SoapPageableLookup<Campaign, CampaignSearchCriteria> soapLookup =
        new SoapPageableLookup<Campaign, CampaignSearchCriteria>(campaignIds,
            new CampaignSoapLookupCallback());

    SimpleDateFormat dateFormat = new SimpleDateFormat("MM-dd-yyyy");
    while (soapLookup.hasNext()) {
      Campaign result = soapLookup.next();
      campaignDefaultLandingPageMap.put(result.getId(), result.getDefaultLandingPageId());
      if (campaignMatchTable != null) {
        campaignMatchTable.write(
            Long.toString(result.getAdvertiserId()),
            Long.toString(result.getId()),
            result.getName(),
            dateFormat.format(result.getStartDate().getTime()),
            dateFormat.format(result.getEndDate().getTime()),
            "1",
            result.getBillingInvoiceNotation());
      }
    }
  }

  // Generates all ad lookups and match table rows for all ads in advertiser.
  private void generateAdsInAdvertiserMatchTable(Long advertiserId, Set<Long> campaignIds,
      MatchTable adMatchTable, MatchTable adPageAssignmentsMatchTable,
      MatchTable creativeMatchTable, MatchTable creativeAssignmentMatchTable,
      MatchTable keyvalMatchTable, MatchTable creativeCustomFieldMatchTable,
      MatchTable metEventMatchTable) throws MatchTableFileException, DfaSoapException {
    // Creative IDs
    Set<Long> creativeSet = new HashSet<Long>();
    // Creative Assignments
    Map<AdBase, Set<CreativeAssignment>> adCreativeAssignmentSetMap =
        new HashMap<AdBase, Set<CreativeAssignment>>();
    // Landing Pages
    Map<Long, String> landingPageMap = new HashMap<Long, String>();
    // Met Events
    Map<Long, RichMediaEventBase> metEventMap = new HashMap<Long, RichMediaEventBase>();


    for (Long campaignId : campaignIds) {
      double campaignCountPercent = 100.0f * campaignCount++ / campaignTotal;
      System.out.printf("\r[%.1f%% completed]", campaignCountPercent);
      // SOAP restriction - Can only retrieve ads of one campaign at a time.
      generateAdsInCampaignMatchTable(advertiserId, campaignId, campaignToAdIdsMap.get(campaignId),
          creativeSet, adCreativeAssignmentSetMap, landingPageMap,
          adMatchTable, adPageAssignmentsMatchTable, keyvalMatchTable);
    }

    // Creative table generation
    Map<Long, CreativeRenderingResult> creativeToRenderingMap =
        new HashMap<Long, CreativeRenderingResult>();
    if (creativeMatchTable != null || creativeCustomFieldMatchTable != null ||
        creativeAssignmentMatchTable != null) {
      // Now process all creatives in this campaign for these ads.
      generateCreativeMatchTable(advertiserId, creativeSet, creativeToRenderingMap,
          metEventMap, creativeMatchTable, creativeCustomFieldMatchTable);
    }

    if (creativeAssignmentMatchTable != null) {
      generateCreativeAssignmentMatchTable(advertiserId, adCreativeAssignmentSetMap,
          creativeToRenderingMap, landingPageMap, creativeAssignmentMatchTable);
    }

    if (metEventMatchTable != null) {
      Map<String, String> richMediaEventTypes = new HashMap<String, String>();
      richMediaEventTypes.put("Timer", "1");
      richMediaEventTypes.put("Exit", "2");
      richMediaEventTypes.put("Counter", "3");

      for (RichMediaEventBase richMediaEventBase : metEventMap.values()) {
        metEventMatchTable.write(
            Long.toString(advertiserId),
            Long.toString(richMediaEventBase.getId()),
            richMediaEventBase.getName(),
            richMediaEventTypes.get(richMediaEventBase.getType()),
            richMediaEventBase.getType());
      }
    }
  }

  // Generates all ad lookups and match table rows for all ads in one campaign.
  private void generateAdsInCampaignMatchTable(Long advertiserId, Long campaignId,
      Set<Long> adIds, Set<Long> creativeSet,
      Map<AdBase, Set<CreativeAssignment>> adCreativeAssignmentSetMap,
      Map<Long, String> landingPageMap,
      MatchTable adMatchTable, MatchTable adPageAssignmentsMatchTable, MatchTable keyvalMatchTable)
          throws MatchTableFileException, DfaSoapException {

    // From iterating through ads store...
    // Sizes
    Set<Long> newInThisCampaignSizes = new HashSet<Long>();
    // Ads themselves
    List<AdBase> adList = new ArrayList<AdBase>();

    // Create LandingPage lookup map
    SoapLookup<LandingPage> landingPageSoapLookup = new SoapLookup<LandingPage>(
        new LandingPageSoapLookupCallback(campaignId));
    while (landingPageSoapLookup.hasNext()) {
      LandingPage result = landingPageSoapLookup.next();
      landingPageMap.put(result.getId(), result.getUrl());
    }

    SoapPageableLookup<AdBase, AdSearchCriteria> adSoapLookup =
        new SoapPageableLookup<AdBase, AdSearchCriteria>(adIds,
            new AdSoapLookupCallback(campaignId));
    while (adSoapLookup.hasNext()) {
      AdBase result = adSoapLookup.next();
      long sizeId = result.getSizeId();
      if (sizeId > 0 && !sizeMap.containsKey(sizeId)) {
        sizeMap.put(sizeId, "");
        newInThisCampaignSizes.add(sizeId);
      }
      adList.add(result);
    }

    // Secondary lookup of sizes.
    if (!newInThisCampaignSizes.isEmpty()) {
      SoapLookup<Size> sizeSoapLookup = new SoapLookup<Size>(newInThisCampaignSizes,
          new SizeSoapLookupCallback());
      while (sizeSoapLookup.hasNext()) {
        Size result = sizeSoapLookup.next();
        String sizeString = result.getWidth() + "x" + result.getHeight();
        sizeMap.put(result.getId(), sizeString);
      }
    }

    for (AdBase result : adList) {
      String adId = Long.toString(result.getId());
      Set<CreativeAssignment> creativeAssignmentSet = new HashSet<CreativeAssignment>();
      extractCreatives(result, creativeAssignmentSet, creativeSet);
      adCreativeAssignmentSetMap.put(result, creativeAssignmentSet);
      if (adMatchTable != null) {
        adMatchTable.write(
            Long.toString(advertiserId),
            Long.toString(campaignId),
            adId,
            result.getName(),
            getClickThruURL(result, landingPageMap,
                campaignDefaultLandingPageMap.get(result.getCampaignId())),
            adTypeMap.get(result.getTypeId()),
            sizeMap.get(result.getSizeId()),
            result.getComments());
      }
      PlacementAssignment[] placementAssignments = result.getPlacementAssignments();
      if (adPageAssignmentsMatchTable != null && placementAssignments != null) {
        for (PlacementAssignment placementAssignment : placementAssignments) {
          adPageAssignmentsMatchTable.write(
              adId,
              Long.toString(placementAssignment.getPlacementId()));
        }
      }
      if (keyvalMatchTable != null) {
        String expression = extractKeywordExpression(result);
        if (expression != null) {
          keyvalMatchTable.write(
              adId,
              expression);
        }
      }
    }

    if (adMatchTable != null) {
      adMatchTable.flush();
    }
    if (keyvalMatchTable != null) {
      keyvalMatchTable.flush();
    }
    if (adPageAssignmentsMatchTable != null) {
      adPageAssignmentsMatchTable.flush();
    }
  }

  // Generate creative assignment match table for all ads in this advertiser/campaign.
  private void generateCreativeAssignmentMatchTable(Long advertiserId,
      Map<AdBase, Set<CreativeAssignment>> creativeAssignmentMap,
      Map<Long, CreativeRenderingResult> creativeToRenderingMap,
      Map<Long, String> landingPageMap,
      MatchTable creativeAssignmentMatchTable) throws MatchTableFileException, DfaSoapException {

    Map<Long, String> creativeGroupMap = new HashMap<Long, String>();
    for (Map.Entry<AdBase, Set<CreativeAssignment>> entry : creativeAssignmentMap.entrySet()) {
      for (CreativeAssignment creativeAssignment : entry.getValue()) {
        long creativeGroupId =
            creativeAssignment.getCreativeGroupAssignment().getCreativeGroup1Id();
        if (creativeGroupId > 0) {
          creativeGroupMap.put(creativeGroupId, "");
        }
        creativeGroupId =
            creativeAssignment.getCreativeGroupAssignment().getCreativeGroup2Id();
        if (creativeGroupId > 0) {
          creativeGroupMap.put(creativeGroupId, "");
        }
      }
    }

    if (!creativeGroupMap.isEmpty()) {
      SoapLookup<CreativeGroup> creativeGroupSoapLookup =
          new SoapLookup<CreativeGroup>(creativeGroupMap.keySet(),
              new CreativeGroupSoapLookupCallback(advertiserId));
      while (creativeGroupSoapLookup.hasNext()) {
        CreativeGroup result = creativeGroupSoapLookup.next();
        creativeGroupMap.put(result.getId(), result.getName());
      }
    }

    for (Map.Entry<AdBase, Set<CreativeAssignment>> entry : creativeAssignmentMap.entrySet()) {
      AdBase ad = entry.getKey();
      String rotationType = "";
      if (ad instanceof RotationGroup) {
        long rotationTypeId = ((RotationGroup) ad).getRotationType();
        for (CreativeRotationType creativeRotationType : CreativeRotationType.values()) {
          if (creativeRotationType.getId() == rotationTypeId) {
            rotationType = creativeRotationType.getName();
            break;
          }
        }
      }
      SimpleDateFormat dateFormat = new SimpleDateFormat("MM-dd-yyyy HH:mm:ss");
      for (CreativeAssignment creativeAssignment : entry.getValue()) {
        CreativeRenderingResult creativeRenderingResult =
            creativeToRenderingMap.get(creativeAssignment.getCreativeId());
        // Only output creative-ad-assignments whose rendering ID is in the DT file
        if (creativeRenderingResult != null) {
          String startDate = "";
          String endDate = "";
          if (creativeAssignment.getStartDate() != null) {
            startDate = dateFormat.format(creativeAssignment.getStartDate().getTime());
          } else if (ad.getStartTime() != null) {
            startDate = dateFormat.format(ad.getStartTime().getTime());
          }
          if (creativeAssignment.getEndDate() != null) {
            endDate = dateFormat.format(creativeAssignment.getEndDate().getTime());
          } else if (ad.getEndTime() != null) {
            endDate = dateFormat.format(ad.getEndTime().getTime());
          }
          creativeAssignmentMatchTable.write(
              Long.toString(ad.getId()),
              Long.toString(creativeRenderingResult.getRenderingId()),
              startDate,
              endDate,
              rotationType,
              creativeGroupMap.get(
                  creativeAssignment.getCreativeGroupAssignment().getCreativeGroup1Id()),
              creativeGroupMap.get(
                  creativeAssignment.getCreativeGroupAssignment().getCreativeGroup2Id()),
              computeClickThroughURL(creativeAssignment.getClickThroughUrl(), landingPageMap,
                  campaignDefaultLandingPageMap.get(ad.getCampaignId())));
        }
      }
    }
    if (creativeAssignmentMatchTable != null) {
      creativeAssignmentMatchTable.flush();
    }
  }

  // Generate all creative match tables for this advertiser and campaign.
  private void generateCreativeMatchTable(Long advertiserId,
      Set<Long> creativeSet, Map<Long, CreativeRenderingResult> creativeToRenderingMap,
      Map<Long, RichMediaEventBase> metEventMap, MatchTable creativeMatchTable,
      MatchTable creativeCustomFieldMatchTable) throws MatchTableFileException, DfaSoapException {

    // Since we only have the rendering ID in the DT file, we need to retrieve all renderings for
    // all the ad in the DT file and match them up with the rendering IDs listed in the DT file.
    // Those creative that have matching rendering ids, we will lookup the creative ID from
    // the SOAP API
    SoapLookup<CreativeRenderingResult> creativeRenderingSoapLookup =
        new SoapLookup<CreativeRenderingResult>(creativeSet,
            new CreativeRenderingSoapLookupCallback());

    Set<Long> foundCreateiveIds = new HashSet<Long>();
    while  (creativeRenderingSoapLookup.hasNext()) {
      CreativeRenderingResult creativeRenderingResult = creativeRenderingSoapLookup.next();
      if (renderingIdsSet.contains(creativeRenderingResult.getRenderingId())) {
        creativeToRenderingMap.put(
            creativeRenderingResult.getCreativeId(), creativeRenderingResult);
        foundCreateiveIds.add(creativeRenderingResult.getCreativeId());
      }
    }

    SoapPageableLookup<CreativeBase, CreativeSearchCriteria> soapLookup =
        new SoapPageableLookup<CreativeBase, CreativeSearchCriteria>(foundCreateiveIds,
            new CreativeSoapLookupCallback(advertiserId));

    Map<Long, CreativeField> creativeFields = new HashMap<Long, CreativeField>();
    Map<Long, String> creativeFieldValues = new HashMap<Long, String>();
    Map<Long, CreativeFieldAssignment[]> creativeFieldAssignmentMap =
        new HashMap<Long, CreativeFieldAssignment[]>();

    while (soapLookup.hasNext()) {
      CreativeBase result  = soapLookup.next();
      if (result == null) {
        continue;
      }
      CreativeFieldAssignment[] creativeFieldAssignments = result.getCreativeFieldAssignments();
      if (creativeFieldAssignments != null && creativeFieldAssignments.length > 0) {
        creativeFieldAssignmentMap.put(result.getId(), creativeFieldAssignments);
        for (CreativeFieldAssignment creativeFieldAssignment : creativeFieldAssignments) {
          creativeFields.put(creativeFieldAssignment.getCreativeFieldId(), null);
          creativeFieldValues.put(creativeFieldAssignment.getCreativeFieldValueId(), null);
        }
      }

      extractMetEvents(advertiserId, result, metEventMap);

      SoapCreativeType soapCreativeType = SoapCreativeType.getById(result.getTypeId());
      String renderingId =
          Long.toString(creativeToRenderingMap.get(result.getId()).getRenderingId());

      // store custom field IDs
      if (creativeMatchTable != null) {
        creativeMatchTable.write(
            Long.toString(advertiserId),
            renderingId,
            Long.toString(result.getId()),
            result.getName(),
            soapCreativeType != null ?
                soapCreativeType.getMatchTableCreativeType().getName() : "",
            soapCreativeType != null ?
                soapCreativeType.getMatchTableCreativeSubType().getName() : "",
            sizeMap.get(result.getSizeId()),
            extractImageURL(result),
            Long.toString(creativeToRenderingMap.get(result.getId()).getRenderingVersion()));
      }
    }

    // Look up custom fields and generate match table for them.
    if (creativeCustomFieldMatchTable != null) {
      SoapPageableLookup<CreativeField, CreativeFieldSearchCriteria> creativeFieldSoapLookup =
          new SoapPageableLookup<CreativeField, CreativeFieldSearchCriteria>(
              creativeFields.keySet(), new CreativeFieldSoapLookupCallback());
      while (creativeFieldSoapLookup.hasNext()) {
        CreativeField result = creativeFieldSoapLookup.next();
        creativeFields.put(result.getId(), result);
      }

      SoapPageableLookup<CreativeFieldValue, CreativeFieldValueSearchCriteria>
      creativeFieldValueSoapLookup =
        new SoapPageableLookup<CreativeFieldValue, CreativeFieldValueSearchCriteria>(
              creativeFieldValues.keySet(), new CreativeFieldValueSoapLookupCallback());
      while (creativeFieldValueSoapLookup.hasNext()) {
        CreativeFieldValue result = creativeFieldValueSoapLookup.next();
        creativeFieldValues.put(result.getId(), result.getName());
      }

      for (Map.Entry<Long, CreativeFieldAssignment[]> entry :
        creativeFieldAssignmentMap.entrySet()) {
        for (CreativeFieldAssignment assignment : entry.getValue()) {
          String renderingId =
              Long.toString(creativeToRenderingMap.get(entry.getKey()).getRenderingId());
          CreativeField creativeField =
              creativeFields.get(assignment.getCreativeFieldId());
          String creativeFieldValue =
              creativeFieldValues.get(assignment.getCreativeFieldValueId());
          creativeCustomFieldMatchTable.write(
              Long.toString(creativeField.getAdvertiserId()),
              renderingId,
              creativeField.getName(),
              creativeFieldValue);
        }
      }
    }

    // Flush the match tables.
    if (creativeMatchTable != null) {
      creativeMatchTable.flush();
    }

    if (creativeCustomFieldMatchTable != null) {
      creativeCustomFieldMatchTable.flush();
    }
  }

  // Utility method to extract click-thru URL property of Ad.
  private String getClickThruURL(AdBase ad, Map<Long, String> landingPageMap,
      Long campaignDefaultLandingPageId) {
    if (ad instanceof ClickTracker) {
      return computeClickThroughURL(((ClickTracker) ad).getClickThroughUrl(), landingPageMap,
          campaignDefaultLandingPageId);
    }
    return null;
  }

  // Actual computation of click-thru URL property of ClickThroughUrl.
  private String computeClickThroughURL(ClickThroughUrl url, Map<Long, String> landingPageMap,
      Long campaignDefaultLandingPageId) {
    if (url.isDefaultLandingPageUsed()) {
      return landingPageMap.get(campaignDefaultLandingPageId);
    } else if (url.getLandingPageId() > 0) {
      return landingPageMap.get(url.getLandingPageId());
    }
    return url.getCustomClickThroughUrl();
  }

  // Utility method to retrieve creative IDs from Ad by trying to downcast object to Ad
  // subclass that supports creative or creativeAssignment property.
  private void extractCreatives(AdBase ad,
      Set<CreativeAssignment> creativeAssignmentSet,
      Set<Long> creativeSet) {
    CreativeAssignment[] creativeAssignmentArray = null;
    if (ad instanceof DefaultAd) {
      creativeAssignmentArray = new CreativeAssignment[1];
      creativeAssignmentArray[0] = ((DefaultAd) ad).getCreativeAssignment();
    } else if (ad instanceof CreativeAd) {
      CreativeAd creativeAd = (CreativeAd) ad;
      creativeAssignmentArray = new CreativeAssignment[1];
      CreativeAssignment dummyCreativeAssignment = new CreativeAssignment();
      dummyCreativeAssignment.setCreativeId(creativeAd.getCreativeId());
      dummyCreativeAssignment.setCreativeGroupAssignment(creativeAd.getCreativeGroupAssignment());
      dummyCreativeAssignment.setClickThroughUrl(creativeAd.getClickThroughUrl());
      dummyCreativeAssignment.setEndDate(creativeAd.getEndTime());
      dummyCreativeAssignment.setStartDate(creativeAd.getStartTime());
      creativeAssignmentArray[0] = dummyCreativeAssignment;
    } else if (ad instanceof MobileAd) {
      creativeAssignmentArray = ((MobileAd) ad).getCreativeAssignments();
    } else if (ad instanceof RotationGroup) {
      creativeAssignmentArray = ((RotationGroup) ad).getCreativeAssignments();
    } else if (ad instanceof TrackingAd) {
      creativeAssignmentArray = ((TrackingAd) ad).getCreativeAssignments();
    }
    if (creativeAssignmentArray != null) {
      for (CreativeAssignment creativeAssignment : creativeAssignmentArray) {
        creativeAssignmentSet.add(creativeAssignment);
        creativeSet.add(creativeAssignment.getCreativeId());
      }
    }
  }

  // Extract keyword expression from Ad if downcasted type supports that property.
  private String extractKeywordExpression(AdBase ad) {
    if (ad instanceof TargetableAdBase) {
      return ((TargetableAdBase) ad).getKeywordExpression();
    }
    if (ad instanceof MobileAd) {
      return ((MobileAd) ad).getKeywordExpression();
    }
    return null;
  }

  // Extract RM events from creative if downcasted Creative type supports that property.
  private void extractMetEvents(Long advertiserId, CreativeBase creative,
      Map<Long, RichMediaEventBase> metEventMap) {
    Set<String> metIdSet = advertiserToMetEventIdMap.get(Long.toString(advertiserId));
    if (creative instanceof RichMediaCreativeBase && metIdSet != null) {
      RichMediaCreativeBase richMediaCreativeBase = (RichMediaCreativeBase) creative;
      for (RichMediaEventBase richMediaEvent : richMediaCreativeBase.getCounterEvents()) {
        if (metIdSet.contains(Long.toString(richMediaEvent.getId()))) {
          metEventMap.put(richMediaEvent.getId(), richMediaEvent);
        }
      }
      for (RichMediaEventBase richMediaEvent : richMediaCreativeBase.getExitEvents()) {
        if (metIdSet.contains(Long.toString(richMediaEvent.getId()))) {
          metEventMap.put(richMediaEvent.getId(), richMediaEvent);
        }
      }
      for (RichMediaEventBase richMediaEvent : richMediaCreativeBase.getTimerEvent()) {
        if (metIdSet.contains(Long.toString(richMediaEvent.getId()))) {
          metEventMap.put(richMediaEvent.getId(), richMediaEvent);
        }
      }
    }
  }

  // Extract image URL property if downcasted Creative type supports that property.
  private String extractImageURL(CreativeBase creative) {
    if (creative instanceof ImageCreativeBase) {
      return ((ImageCreativeBase) creative).getAssetFilename();
    } else if (creative instanceof RedirectCreativeBase) {
      return ((RedirectCreativeBase) creative).getRedirectUrl();
    }
    return null;
  }

  // SoapLookupCallbacks

  private class SizeSoapLookupCallback implements SoapLookupCallback<Size> {

    public int getRecords(long[] ids, List<Size> recordList)
        throws ApiException, RemoteException {

      SizeSearchCriteria sizeSearchCriteria = new SizeSearchCriteria();
      if (ids != null) {
        sizeSearchCriteria.setIds(ids);
      }
      sizeSearchCriteria.setWidth(-1);
      sizeSearchCriteria.setHeight(-1);
      SizeRecordSet sizeRecordSet = sizeService.getSizes(sizeSearchCriteria);
      Size[] sizeArray = sizeRecordSet.getRecords();
      for (Size size : sizeArray) {
        recordList.add(size);
      }
      return sizeArray.length;
    }
  }

  private class AdSoapLookupCallback
    implements SoapPageableLookupCallback<AdBase, AdSearchCriteria> {

    AdSoapLookupCallback(long campaignId) {
      this.campaignId = campaignId;
    }

    public int getRecords(AdSearchCriteria criteria, List<AdBase> recordList)
        throws ApiException, RemoteException {

      // Lazy initialization
      if (adTypeMap.isEmpty()) {
        AdType[] adTypes = adService.getAdTypes();
        // Display ad type and its ID.
        for (AdType result : adTypes) {
          adTypeMap.put(result.getId(), result.getName());
        }
      }

      AdRecordSet adRecordSet = adService.getAds(criteria);
      for (AdBase ad : adRecordSet.getRecords()) {
        recordList.add(ad);
      }
      return adRecordSet.getTotalNumberOfPages();
    }

    public AdSearchCriteria getCriteria() {
      AdSearchCriteria adSearchCriteria = new AdSearchCriteria();
      adSearchCriteria.setCampaignId(campaignId);
      return adSearchCriteria;
    }

    private long campaignId;
  }

  private class CreativeSoapLookupCallback
    implements SoapPageableLookupCallback<CreativeBase, CreativeSearchCriteria> {

    CreativeSoapLookupCallback(long advertiserId) {
      this.advertiserId = advertiserId;
    }

    public int getRecords(CreativeSearchCriteria criteria, List<CreativeBase> recordList)
        throws ApiException, RemoteException {

      CreativeRecordSet creativeRecordSet = creativeService.getCreatives(criteria);
      for (CreativeBase entity : creativeRecordSet.getRecords()) {
        recordList.add(entity);
      }
      return creativeRecordSet.getTotalNumberOfPages();
    }

    public CreativeSearchCriteria getCriteria() {
      CreativeSearchCriteria creativeSearchCriteria = new CreativeSearchCriteria();
      creativeSearchCriteria.setAdvertiserId(advertiserId);
      return creativeSearchCriteria;
    }

    private long advertiserId;
  }

  private class CreativeFieldSoapLookupCallback
    implements SoapPageableLookupCallback<CreativeField, CreativeFieldSearchCriteria> {

    CreativeFieldSoapLookupCallback() {
    }

    public int getRecords(CreativeFieldSearchCriteria criteria, List<CreativeField> recordList)
        throws ApiException, RemoteException {
      CreativeFieldRecordSet creativeFieldRecordSet =
          creativeFieldeService.getCreativeFields(criteria);
      for (CreativeField entity : creativeFieldRecordSet.getRecords()) {
        recordList.add(entity);
      }
      return creativeFieldRecordSet.getTotalNumberOfPages();
    }

    public CreativeFieldSearchCriteria getCriteria() {
      CreativeFieldSearchCriteria searchCriteria = new CreativeFieldSearchCriteria();
      return searchCriteria;
    }
  }

  private class CreativeFieldValueSoapLookupCallback
    implements SoapPageableLookupCallback<CreativeFieldValue, CreativeFieldValueSearchCriteria> {

    CreativeFieldValueSoapLookupCallback() {
    }

    public int getRecords(CreativeFieldValueSearchCriteria criteria,
        List<CreativeFieldValue> recordList)
        throws ApiException, RemoteException {
      CreativeFieldValueRecordSet recordSet =
          creativeFieldeService.getCreativeFieldValues(criteria);
      for (CreativeFieldValue entity : recordSet.getRecords()) {
        recordList.add(entity);
      }
      return recordSet.getTotalNumberOfPages();
    }

    public CreativeFieldValueSearchCriteria getCriteria() {
      CreativeFieldValueSearchCriteria searchCriteria = new CreativeFieldValueSearchCriteria();
      return searchCriteria;
    }
  }

  private class CreativeGroupSoapLookupCallback implements SoapLookupCallback<CreativeGroup> {

    private long[] advertiserIds;

    CreativeGroupSoapLookupCallback(Long advertiserId) {
      this.advertiserIds = new long[1];
      advertiserIds[0] = advertiserId;
    }

    public int getRecords(long[] ids, List<CreativeGroup> recordList)
        throws ApiException, RemoteException {

      CreativeGroupSearchCriteria searchCriteria = new CreativeGroupSearchCriteria();
      searchCriteria.setIds(ids);
      searchCriteria.setAdvertiserIds(advertiserIds);

      CreativeGroupRecordSet recordSet =
          creativeGroupService.getCreativeGroups(searchCriteria);
      CreativeGroup[] creativeGroupArray = recordSet.getRecords();
      for (CreativeGroup entity : creativeGroupArray) {
        recordList.add(entity);
      }
      return creativeGroupArray.length;
    }
  }

  private class CampaignSoapLookupCallback implements
    SoapPageableLookupCallback<Campaign, CampaignSearchCriteria> {

    public int getRecords(CampaignSearchCriteria criteria, List<Campaign> recordList)
        throws ApiException, RemoteException {
      CampaignRecordSet records = campaignService.getCampaignsByCriteria(criteria);
      int pages = records.getTotalNumberOfPages();
      for (Campaign record : records.getRecords()) {
        recordList.add(record);
      }
      return pages;
    }

    public CampaignSearchCriteria getCriteria() {
      return new CampaignSearchCriteria();
    }
  }

  private class LandingPageSoapLookupCallback implements SoapLookupCallback<LandingPage> {

    private long campaignId;
    LandingPageSoapLookupCallback(long campaignId) {
      this.campaignId = campaignId;
    }

    public int getRecords(long[] ids, List<LandingPage> recordList)
        throws ApiException, RemoteException {
      if (campaignId > 0) {
        LandingPageRecordSet landingPages = campaignService.getLandingPagesForCampaign(campaignId);
        LandingPage[] landingPageArray = landingPages.getRecords();
        for (LandingPage landingPage : landingPageArray) {
          recordList.add(landingPage);
        }
        return landingPageArray.length;
      }
      return 0;
    }
  }

  private class CreativeRenderingSoapLookupCallback
    implements SoapLookupCallback<CreativeRenderingResult> {

    public int getRecords(long[] ids, List<CreativeRenderingResult> recordList)
        throws ApiException, RemoteException {
      CreativeRenderingRequest creativeRenderingRequest = new CreativeRenderingRequest();
      creativeRenderingRequest.setCreativeIds(ids);
      CreativeRenderingResult[] creativeRenderingResults =
          creativeService.getCreativeRenderings(creativeRenderingRequest);
      for (CreativeRenderingResult creativeRenderingResult : creativeRenderingResults) {
        recordList.add(creativeRenderingResult);
      }
      return creativeRenderingResults.length;
    }
  }
}
